Chapter 4: Securing the Front Door: Protecting Agents and Execution Entry Points

Every language model invocation incurs real costs in dollar spend, API rate limits, and processing latency. Allowing an unauthorized request past the perimeter consumes tokens formatting prompts, retrieving vector documents, and calling models, only to fail at a downstream tool check. Securing the front door means placing a Policy Enforcement Point (PEP: an inline boundary mechanism that intercepts requests to enforce access decisions) directly at the agent entry point, querying a Policy Decision Point (PDP: the centralized engine that evaluates access rules and returns authorization verdicts) before prompts are rendered or model endpoints are reached.

The Zero-Token Denial Principle

The Zero-Token Denial Principle states that if a caller lacks the entitlement to invoke an agent or execution graph, exactly zero model tokens must be consumed evaluating that refusal.

Relying on system prompt instructions (e.g., "Decline if the user is not an administrator") violates this principle:

• Economic and Latency Waste: Rejections still incur serialization overhead, HTTP latency, and input/output token billing.
• Privilege Escalation Exposure: Unhardened entry points allow unauthenticated or unauthorized callers into the model execution loop, relying purely on conversational steering.

Security Boundary Distinction: The Entry PEP enforces caller identity entitlement (verifying roles, clearance tier, and tenant isolation before invocation). Content-level prompt inspection (e.g. neutralizing jailbreak overrides like "IGNORE ALL PREVIOUS INSTRUCTIONS") is handled upstream by the Ingestion Admission Control Scanner (covered in Chapter 5) and input payload sanitizers (Chapter 9). SecureRunnable intercepts the request completely outside the language model loop, terminating unauthorized callers in sub-millisecond time.

SecureRunnable Interception Lifecycle

Because LangChain agents and chains adhere to LCEL (LangChain Expression Language: a declarative syntax used to compose LangChain primitives into executable chains), SecureRunnable acts as a non-invasive transparent proxy around any Runnable or callable. Upon invocation (invoke, ainvoke, stream), it executes a five-step sequence:

1. Identity Extraction & Resolution: Resolves SecurityIdentity from config["configurable"]["security_identity"] via IdentityBridge. Missing identities default to anonymous and fail closed.
2. In-Band Config & Callback Preparation: Automatically injects CONFIG_KEY_SECURITY_IDENTITY, CONFIG_KEY_INVOCATION_PARAMS, and the active sec:pdp_engine into config["configurable"], while appending a SecureCallbackHandler to config["callbacks"] so downstream tool and RAG operations are transparently intercepted without manual callback wiring.
3. PDP Authorization Evaluation: Submits a standardized PolicyEvaluationRequest tuple (subject, action, resource, context) to the PDP engine to enforce tenant isolation, required roles, and minimum clearance floors. Denials record structured audit events and raise an unchecked AgentAccessDeniedException.
4. Ambient Context Binding: Sets the verified identity into SecurityContextHolder (with a try/finally block guaranteeing context reset), ensuring thread-local and async task propagation.
5. Execution & Outbound Boundary: Invokes the underlying runnable, applying optional in-flight PII redaction and declarative violation fallback handling.

The library implements this lifecycle by wrapping the target runnable in an inline evaluation gate that manages ambient context and policy enforcement:

langchain_security_extension/security_adapter_langchain/runnable_guard.py

class SecureRunnable:
    """Transparent proxy enforcing PDP entry authorization prior to runnable execution."""

    def invoke(self, input_data: Any, *args: Any, **kwargs: Any) -> Any:
        config = kwargs.get("config", args[0] if len(args) > 0 else None)

        # 1. Gate entry: evaluate roles, clearance, and tenant against PDP (raises if denied)
        identity = self._evaluate_entry_access(config)

        # 2. Bind ambient caller identity to execution context
        token = SecurityContextHolder.set_identity(identity)
        try:
            # 3. Transparently pass execution to underlying Runnable or agent chain
            return self._runnable.invoke(input_data, config=config)
        finally:
            # 4. Guarantee context cleanup to prevent cross-request leakage
            SecurityContextHolder.reset_identity(token)

Code Hardening: The Decoupled PEP Pattern

To harden your enterprise application, decouple your public agent entry methods from underlying execution logic using the Decoupled PEP Pattern. The public entrypoint acts as the Policy Enforcement Point (PEP) gateway—validating caller RunnableConfig and delegating to self.secured_workflow.invoke()—guaranteeing that LLM prompts, token generation, and downstream tools are only ever reached if the PDP grants permission.

In our reference assistant, this pattern is implemented in app/agents/triage_agent.py across two focused steps:

Step 1: Wrapping Business Logic with SecureRunnable

During agent initialization, wrap your core execution method with SecureRunnable.wrap(), declaring the agent's target name, required roles, minimum clearance, and permitted tenant boundaries:

app/agents/triage_agent.py

@@ -47,6 +59,13 @@ class TriageAgent:
             self.model_class,
             self.api_key
         )
+        self.secured_workflow = SecureRunnable.wrap(
+            runnable=self._execute_triage,
+            pdp_engine=get_pdp_engine(),
+            agent_name="triage_agent",
+            required_roles=TRIAGE_REQUIRED_ROLES,
+            min_clearance=TRIAGE_MIN_CLEARANCE,
+            target_tenant=ALLOWED_TENANTS,
+        )

Explanation: Instead of calling the LLM directly, self.secured_workflow proxies calls to _execute_triage. Before _execute_triage is invoked, SecureRunnable queries the PDP to verify caller clearance, tenant matching, and roles.

Step 2: Guarding the Public Entry Point

Next, update the public entry method to extract or build RunnableConfig, delegating execution to the secured wrapper:

app/agents/triage_agent.py

@@ -126,8 +130,10 @@ class TriageAgent:
     def triage(self, request: TicketRequest, *args: Any, **kwargs: Any) -> IncidentContext:
+        run_config: Optional[Dict[str, Any]] = args[0] if len(args) > 0 else kwargs.get("config")
+        if run_config is None:
+            sec_identity = security_identity_from_context(request.securityContext)
+            run_config = IdentityBridge.create_runnable_config(sec_identity)
+        return self.secured_workflow.invoke(request, config=run_config)

Explanation: The public triage() method ensures caller identity is bundled into RunnableConfig. If the caller lacks authorization, self.secured_workflow.invoke() immediately raises AgentAccessDeniedException or TenantIsolationError before _execute_triage begins formatting prompts. For the full implementation, see app/agents/triage_agent.py in the reference assistant.

Step 3: Handling Entry Denials at the API Transport Boundary

At your outer HTTP router or controller, catch entry exceptions to construct standardized error responses, terminating unauthorized attempts with zero model token spend:

app/routers/tickets.py

@@ -48,8 +54,18 @@ def submit_ticket(
     sec_identity = security_identity_from_context(sec_context)
     run_config = IdentityBridge.create_runnable_config(sec_identity)
 
-    # Triage Phase
-    incident_context = triage_agent_instance.triage(request)
+    # Triage Phase (Guarded by SecureRunnable Entry PEP)
+    try:
+        incident_context = triage_agent_instance.triage(request, config=run_config)
+    except TenantIsolationError as exc:
+        logger.warning(f"Triage entry tenant isolation violation: {exc}")
+        return build_entry_access_denied_report(request, exc, response, "BLOCKED_TENANT")
+    except AgentAccessDeniedException as exc:
+        logger.warning(f"Triage entry access denied: {exc}")
+        return build_entry_access_denied_report(request, exc, response, "BLOCKED")
@@ -87,7 +92,12 @@ def submit_ticket(
     # Notification Dispatch
     notification_service_instance.emit_resolution_notification(report)
 
-    if report.stageResults and report.stageResults.get("remediation_security") in ("BLOCKED_POLICY", "BLOCKED_TENANT", "BLOCKED"):
+    if report.stageResults and report.stageResults.get("remediation_security") in (
+        "BLOCKED_POLICY",
+        "BLOCKED_TENANT",
+        "BLOCKED",
+        "BLOCKED_CIRCUIT_BREAKER",
+    ):
+        if response:
+            response.status_code = status.HTTP_403_FORBIDDEN
+        return report

Explanation: Where possible, execution of the agent LLM call is blocked upfront at the entry PEP (e.g., for unauthorized roles or tenant mismatches), terminating the request with zero model token spend. Whatever slips past entry checks because it requires runtime evaluation is blocked downstream before the tool call executes, ensuring critical mutative operations cannot run without authority. The resolution notification is dispatched for full audit traceability, and the endpoint halts with HTTP 403 Forbidden.

Multi-Agent Delegation & Context-Preserving Handoff

In enterprise multi-agent workflows, an initial triage agent frequently delegates downstream tasks to specialized execution agents (such as a RemediationAgent). When transitioning between agents, naively invoking the second agent can drop caller identity, invalidate L1 decision cache entries, or lose OpenTelemetry trace IDs.

SecureRunnable.handoff() (and asynchronous ahandoff()) solves this by propagating in-band RunnableConfig across agent boundaries while marking sec:handoff_propagated = True and verifying the target agent's entry entitlements:

app/agents/remediation_agent.py

@@ -133,13 +93,13 @@ class RemediationAgent:
     def process_incident(self, context: IncidentContext, *args: Any, **kwargs: Any) -> RemediationReport:
-        config: Optional[Dict[str, Any]] = args[0] if len(args) > 0 else kwargs.get("config")
-        return self._execute_remediation(context, config=config)
+        run_config: Optional[Dict[str, Any]] = args[0] if len(args) > 0 else kwargs.get("config")
+        if run_config is None:
+            sec_identity = security_identity_from_context(context.securityContext)
+            run_config = IdentityBridge.create_runnable_config(sec_identity)
+
+        return SecureRunnable.handoff(self.secured_workflow, context, run_config)

Explanation: SecureRunnable.handoff() carries the caller's verified SecurityIdentity and trace attributes directly to the secondary agent, ensuring consistent authorization enforcement without global state. See app/agents/remediation_agent.py in the reference assistant.

Declarative Violation Handler & Error Boundary

When an agent executes tools or secondary runnables within an LCEL graph, policy rejections from within the execution graph shouldn't trigger unhandled application crashes or expose internal stack traces. In your application, bind an error boundary to SecureRunnable using .with_violation_handler(violation_handler, fallback_builder):

app/agents/remediation_agent.py

@@ -96,8 +96,16 @@ class RemediationAgent:
+        self.violation_handler = ToolViolationHandler.default_handler(get_pdp_engine())
+        self.secured_workflow = SecureRunnable.wrap(
+            runnable=self._execute_remediation,
+            pdp_engine=get_pdp_engine(),
+            agent_name="remediation_agent",
+            required_roles=TRIAGE_REQUIRED_ROLES,
+            min_clearance=TRIAGE_MIN_CLEARANCE,
+            target_tenant=ALLOWED_TENANTS,
+        ).with_violation_handler(
+            violation_handler=self.violation_handler,
+            fallback_builder=build_violation_fallback_report,
+        )

Explanation: When a downstream tool execution is blocked by policy, the error boundary catches the exception, resolves the violation details via the ToolViolationHandler, and invokes the fallback_builder to generate a graceful remediation report containing security diagnostic telemetry (e.g., remediation_security: BLOCKED_POLICY, remediation_execution: ABORTED) rather than aborting abruptly.

Verification Tests

Validate front-door execution guards, identity verification, and multi-agent context propagation using the project's security test suite:

• Test 03: Unauthorized User Request Blocked (03-unauthorized-user-blocked.sh)
  Verifies that an unauthorized caller (e.g., external attacker with ROLE_GUEST) is rejected at the entry PEP with HTTP 403 Forbidden before any LLM prompt formatting or token consumption occurs.
  BASH

  ./basic-security-tests/03-unauthorized-user-blocked.sh

• Test 04: Cross-Tenant Isolation Blocked (04-cross-tenant-isolation-blocked.sh)
  Confirms that requests originating from unauthorized or foreign tenants trigger a TenantIsolationError at the SecureRunnable gate and are blocked with HTTP 403.
  BASH

  ./basic-security-tests/04-cross-tenant-isolation-blocked.sh

• Test 05: Clearance Floor Violation Blocked (05-clearance-floor-violation-blocked.sh)
  Confirms that callers with clearance ranks below the required min_clearance threshold are stopped at the front door by SecureRunnable with an AgentAccessDeniedException.
  BASH

  ./basic-security-tests/05-clearance-floor-violation-blocked.sh

• Test 09: Subagent Async Context Propagation (09-subagent-async-context-propagated.sh)
  Verifies that SecurityIdentity and runnable configuration are seamlessly preserved across asynchronous thread and agent handoff boundaries.
  BASH

  ./basic-security-tests/09-subagent-async-context-propagated.sh

Library unit tests in langchain-security-extension/tests/test_runnable_guard.py further validate the underlying primitives:

pytest langchain-security-extension/tests/test_runnable_guard.py -k "test_secure_runnable"